package com.tn.rtdata.management;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.InvalidDestinationException;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import com.tn.openrpc.serialization.SerializationException;
import com.tn.openrpc.serialization.Serializer;
import com.tn.openrpc.util.Factory;
import com.tn.openrpc.util.FactoryException;
import com.tn.rtdata.DataEvent;
import com.tn.rtdata.FillException;
import com.tn.rtdata.FillIdentity;
import com.tn.rtdata.FillListener;
import com.tn.rtdata.SessionException;

/**
 * A manager service that manages sessions.
 */
public class SessionManager
{
  private static final Logger LOGGER = LoggerFactory.getLogger(SessionManager.class);

  //The connection factory is initialized via the constructor and must be static but because it is ultimately used by the
  //MessageProducingFillListener which, like all FillListeners, could be serialized when a distributed cache is used.
  private static ConnectionFactory connectionFactory;

  private FillManager fillManager;
  private HeartbeatManager heartbeatManager;
  private Serializer<String> serializer;
  private Factory<String> sessionIdFactory;
  private Map<String, Map<FillIdentity, FillListener>> sessionCache;

  /**
   * Creates a new <code>SessionManager</code>.
   *
   * @param sessionIdFactory the factory that creates session ids.
   * @param sessionCache     the cache used to hold session ids.
   * @param heartbeatManager the heartbeat manager.
   */
  public SessionManager(
    ConnectionFactory connectionFactory,
    Factory<String> sessionIdFactory,
    Map<String, Map<FillIdentity, FillListener>> sessionCache,
    FillManager fillManager,
    HeartbeatManager heartbeatManager,
    Serializer<String> serializer
  )
  {
    //noinspection AccessStaticViaInstance
    this.connectionFactory = connectionFactory;
    this.sessionIdFactory = sessionIdFactory;
    this.sessionCache = sessionCache;
    this.fillManager = fillManager;
    this.heartbeatManager = heartbeatManager;
    this.serializer = serializer;
    this.heartbeatManager.addHeartbeatListener(new SessionClosingHeartbeatListener());
  }

  /**
   * Closes the session.
   */
  public void close(String sessionId)
  {
    LOGGER.info("Closing session: {}", sessionId);

    stopFills(sessionCache.get(sessionId));
    sessionCache.remove(sessionId);
    heartbeatManager.remove(sessionId);

    LOGGER.debug("Closed session: {}", sessionId);
  }

  /**
   * Connects a new session.
   */
  public String connect() throws SessionException
  {
    try
    {
      String sessionId = sessionIdFactory.newInstance();

      LOGGER.info("Created session with id: {}", sessionId);

      sessionCache.put(sessionId, new HashMap<FillIdentity, FillListener>());
      heartbeatManager.heartbeat(sessionId);

      return sessionId;
    }
    catch (FactoryException e)
    {
      throw new SessionException("An error occurred creating a session id.", e);
    }
  }

  /**
   * Processes a heartbeat for the session.
   */
  public void heartbeat(String sessionId)
  {
    heartbeatManager.heartbeat(sessionId);
  }

  /**
   * Starts the fill with the specified <code>fillName</code> for the <code>sessionId</code> using the <code>parameters</code>.
   */
  public void startFill(String sessionId, String fillName, List<?> parameters, String fillEventDestination) throws FillException
  {
    LOGGER.debug("Got fill request for session id: {}, fill name: {}, parameters: {}", new Object[]{sessionId, fillName, parameters});

    if (!sessionCache.containsKey(sessionId))
    {
      throw new FillException("Session does not exist: " + sessionId);
    }

    FillIdentity fillIdentity = new FillIdentity(fillName, parameters);
    FillListener fillListener = new MessageProducingFillListener(sessionId, fillEventDestination, serializer);

    sessionCache.get(sessionId).put(fillIdentity, fillListener);
    fillManager.startFill(fillIdentity, fillListener);
  }

  /**
   * Stops the fill with the specified <code>fillName</code> for the <code>sessionId</code> using the <code>parameters</code>.
   */
  public void stopFill(String sessionId, String fillName, List<?> parameters) throws FillException
  {
    if (!sessionCache.containsKey(sessionId))
    {
      throw new FillException("Session does not exist: " + sessionId);
    }

    FillIdentity fillIdentity = new FillIdentity(fillName, parameters);
    stopFillListener(fillIdentity, sessionCache.get(sessionId).remove(fillIdentity));
  }

  /**
   * Stops with the fill with the specified <code>fillIdentity</code> for the <code>fillListener</code>.
   */
  private void stopFillListener(FillIdentity fillIdentity, FillListener fillListener) throws FillException
  {
    fillManager.stopFill(fillIdentity, fillListener);
  }

  /**
   * Stops the fills with the specified <code>fillListeners</code>.
   */
  private void stopFills(Map<FillIdentity, FillListener> fillListeners)
  {
    if (fillListeners != null)
    {
      for (FillIdentity fillIdentity : fillListeners.keySet())
      {
        try
        {
          stopFillListener(fillIdentity, fillListeners.get(fillIdentity));
        }
        catch (FillException e)
        {
          LOGGER.error("An error occurred stopping fill with identity: " + fillIdentity, e);
        }
      }
    }
  }

  /**
   * A specialization of <code>HeartbeatManager.HeartbeatListener</code> that closes the session when a terminate
   * event occurs.
   */
  private class SessionClosingHeartbeatListener extends HeartbeatManager.HeartbeatListener
  {
    /**
     * {@inheritDoc}
     */
    @Override
    public void terminate(String id)
    {
      close(id);
    }
  }

  /**
   * An implementation of <code>FillListener</code> that publishes a <code>Message</code> to a JMS <code>Destination</code>.
   */
  public static class MessageProducingFillListener implements FillListener
  {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageProducingFillListener.class);

    private String dataEventDestination;
    private Serializer<String> serializer;
    private String sessionId;
    private boolean sessionFailed;

    /**
     * Creates a new <code>MessageProducingFillListener</code> initialized with the <code>sessionId</code>,
     * <code>dataEventDestination</code> and <code>serializer</code>.
     */
    public MessageProducingFillListener(String sessionId, String dataEventDestination, Serializer<String> serializer)
    {
      this.sessionId = sessionId;
      this.dataEventDestination = dataEventDestination;
      this.serializer = serializer;
      this.sessionFailed = false;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void fillInitialized(String eventId, Collection fillEntries)
    {
      try
      {
        Collection<DataEvent> dataEvents = new ArrayList<DataEvent>();
        for (Object fillEntry : fillEntries)
        {
          dataEvents.add(new DataEvent<Object>(eventId, DataEvent.Type.INITIAL, fillEntry, null));
        }

        sendDataEventMessage(dataEvents);
      }
      catch (Exception e)
      {
        LOGGER.error("An error occurred sending initialize data events.", e);
      }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void fillReset(String eventId)
    {
      sendDataEventMessage(new DataEvent<Object>(eventId, DataEvent.Type.RESET, null, null));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryAdded(String eventId, Object newEntry)
    {
      try
      {
        sendDataEventMessage(new DataEvent<Object>(eventId, DataEvent.Type.ADD, newEntry, null));
      }
      catch (Exception e)
      {
        LOGGER.error(
          "An error occurred sending initialize data events.",
          e.getCause() instanceof SerializationException ? e.getCause() : e
        );
      }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryDeleted(String eventId, Object oldEntry)
    {
      try
      {
        sendDataEventMessage(new DataEvent<Object>(eventId, DataEvent.Type.DELETE, null, oldEntry));
      }
      catch (Exception e)
      {
        LOGGER.error(
          "An error occurred sending initialize data events.",
          e.getCause() instanceof SerializationException ? e.getCause() : e
        );
      }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryUpdated(String eventId, Object newEntry, Object oldEntry)
    {
      try
      {
        sendDataEventMessage(new DataEvent<Object>(eventId, DataEvent.Type.UPDATE, newEntry, oldEntry));
      }
      catch (Exception e)
      {
        LOGGER.error(
          "An error occurred sending initialize data events.",
          e.getCause() instanceof SerializationException ? e.getCause() : e
        );
      }
    }

    /**
     * Compares this object to another for equality.  For another object to be considered equal to a
     * <code>MessageProducingFillListener</code> the other object must be a <code>MessageProducingFillListener</code> and must have been
     * initialized with the same <code>sessionId</code> and <code>dataEventDestination</code>.
     *
     * @param other the other object.
     *
     * @return <code>true</code> if the objects are considered equal, otherwise <code>false</code>.
     */
    @Override
    public boolean equals(Object other)
    {
      return this == other || other != null && getClass() == other.getClass() &&
      new EqualsBuilder()
        .append(sessionId, ((MessageProducingFillListener)other).sessionId)
        .append(dataEventDestination, ((MessageProducingFillListener)other).dataEventDestination)
        .isEquals();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode()
    {
      return new HashCodeBuilder()
        .append(sessionId)
        .append(dataEventDestination)
        .toHashCode();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString()
    {
      return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
        .append("sessionId", sessionId)
        .append("dataEventDestination", dataEventDestination)
        .toString();
    }

    /**
     * Sends a <code>DataEvent</code> message.
     */
    private void sendDataEventMessage(final Object dataEvent)
    {
      if (!sessionFailed)
      {
        LOGGER.debug("Sending data event(s): {}", dataEvent);

        try
        {
          new JmsTemplate(connectionFactory).send(
            dataEventDestination,
            new MessageCreator()
            {
              @Override
              public Message createMessage(Session session) throws JMSException
              {
                try
                {
                  return session.createTextMessage(serializer.serialize(dataEvent));
                }
                catch (SerializationException e)
                {
                  throw new RuntimeException(e);
                }
              }
            }
          );
        }
        catch (InvalidDestinationException e)
        {
          LOGGER.warn(
            "Destination: {} has been deleted, session: {} assumed to have failed.",
            dataEventDestination,
            sessionId
          );
          sessionFailed = true;
        }
      }
    }
  }
}
